package com.hitqz.robot.dispatch.netty;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.hitqz.robot.api.business.dto.RobotAddDispatchDto;
import com.hitqz.robot.api.dispatch.model.MessageDto;
import com.hitqz.robot.api.dispatch.protocol.ProtocolEnum;
import com.hitqz.robot.api.dispatch.protocol.dto.FormConfigModalDto;
import com.hitqz.robot.api.dispatch.protocol.dto.RobotAsyncCmdDto;
import com.hitqz.robot.api.dispatch.protocol.dto.RobotRegisterDto;
import com.hitqz.robot.biz.websocket.endpoint.SubscriptionEndpoint;
import com.hitqz.robot.common.core.util.R;
import com.hitqz.robot.common.core.util.RedisUtils;
import com.hitqz.robot.dispatch.netty.listener.QueryListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author xupeikun
 */
@Slf4j
@Component
@ChannelHandler.Sharable
public class NettyMessageHandler extends ChannelInboundHandlerAdapter {

	private static final ObjectMapper mapper = new ObjectMapper();

	public static final ThreadPoolExecutor queryPool = new ThreadPoolExecutor(10, 20,
			1L, TimeUnit.SECONDS,
			new LinkedBlockingQueue<Runnable>(20),
			Executors.defaultThreadFactory(),
			new ThreadPoolExecutor.DiscardPolicy());

	/**
     * 客户端连接上来
     * */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
		// 不做逻辑处理
		super.channelActive(ctx);
    }

    /**
     * 客户端发送消息过来
     * */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		super.channelRead(ctx, msg);
        MessageDto dto = (MessageDto) msg;
        String data = new String(dto.getData(), StandardCharsets.UTF_8);
		if(log.isDebugEnabled()) {
			log.debug("[netty]data:{}",data);
		}
		try{
			//分发事件
			ObjectMapper mapper = new ObjectMapper();
			mapper.registerModule(new JavaTimeModule());
			ProtocolEnum protocolEnum = ProtocolEnum.CMD_PROTOCOL_MAP.get(dto.getCmd());
			Object event = mapper.readValue(data, protocolEnum.getClazz());
			if (event instanceof RobotRegisterDto) {
				//
				RobotRegisterDto robotRegisterDto = (RobotRegisterDto) event;
				log.info("机器人注册" + robotRegisterDto.getClientId());
				ChannelCache.add(ctx, robotRegisterDto.getClientId());
				queryPool.execute(()-> {
					try {
						TimeUnit.SECONDS.sleep(5);
					} catch (InterruptedException e) {
						throw new RuntimeException(e);
					}
					RobotAddDispatchDto dispatchDto = new RobotAddDispatchDto();
					dispatchDto.setRobotSn(robotRegisterDto.getClientId());
					SpringUtil.getApplicationContext().publishEvent(dispatchDto);
				});
				queryPool.execute(()-> {
					// 回复成功
					MessageDto messageDto= new MessageDto();
					messageDto.setSn(dto.getSn());
					messageDto.setCmd(ProtocolEnum.ROBOT_REGISTER.getCmd());
					messageDto.setData(JSONObject.toJSONString(R.ok()).getBytes(StandardCharsets.UTF_8));
					messageDto.setClientId(dto.getClientId());
					ChannelCache.send2Client(robotRegisterDto.getClientId(), BeanUtil.copyProperties(messageDto,MessageDto.class));
					//// 下发同步机器人信息
					messageDto.setSn(0);
					messageDto.setCmd(ProtocolEnum.ROBOT_SYNC_CMD.getCmd());
					RobotAsyncCmdDto robotAsyncCmdDto = new RobotAsyncCmdDto();
					robotAsyncCmdDto.setCmd(100);
					messageDto.setData(JSONObject.toJSONString(robotAsyncCmdDto).getBytes(StandardCharsets.UTF_8));
					messageDto.setClientId(dto.getClientId());
					ChannelCache.send2Client(robotRegisterDto.getClientId(), BeanUtil.copyProperties(messageDto,MessageDto.class));
					//// 下发地图
					messageDto.setSn(0);
					robotAsyncCmdDto.setCmd(101);
					messageDto.setCmd(ProtocolEnum.ROBOT_SYNC_CMD.getCmd());
					messageDto.setData(JSONObject.toJSONString(robotAsyncCmdDto).getBytes(StandardCharsets.UTF_8));
					messageDto.setClientId(dto.getClientId());
					ChannelCache.send2Client(robotRegisterDto.getClientId(), BeanUtil.copyProperties(messageDto,MessageDto.class));
					// 同步点位
					messageDto.setSn(0);
					robotAsyncCmdDto.setCmd(102);
					messageDto.setCmd(ProtocolEnum.ROBOT_SYNC_CMD.getCmd());
					messageDto.setData(JSONObject.toJSONString(robotAsyncCmdDto).getBytes(StandardCharsets.UTF_8));
					messageDto.setClientId(dto.getClientId());
					ChannelCache.send2Client(robotRegisterDto.getClientId(), BeanUtil.copyProperties(messageDto,MessageDto.class));
					// 同步工单
					messageDto.setSn(0);
					robotAsyncCmdDto.setCmd(103);
					messageDto.setCmd(ProtocolEnum.ROBOT_SYNC_CMD.getCmd());
					messageDto.setData(JSONObject.toJSONString(robotAsyncCmdDto).getBytes(StandardCharsets.UTF_8));
					messageDto.setClientId(dto.getClientId());
					ChannelCache.send2Client(robotRegisterDto.getClientId(), BeanUtil.copyProperties(messageDto,MessageDto.class));
					// 同步设备
					messageDto.setSn(0);
					robotAsyncCmdDto.setCmd(104);
					messageDto.setCmd(ProtocolEnum.ROBOT_SYNC_CMD.getCmd());
					messageDto.setData(JSONObject.toJSONString(robotAsyncCmdDto).getBytes(StandardCharsets.UTF_8));
					messageDto.setClientId(dto.getClientId());
                    // 105 哨兵模式开 106 关
                    messageDto.setSn(0);
                    Boolean enabled = RedisUtils.get("sentinelMode:" + dto.getClientId());
                    if (enabled != null && enabled) {
                        robotAsyncCmdDto.setCmd(105);
                    }else {
                        robotAsyncCmdDto.setCmd(106);
                    }
                    messageDto.setCmd(ProtocolEnum.ROBOT_SYNC_CMD.getCmd());
                    messageDto.setData(JSONObject.toJSONString(robotAsyncCmdDto).getBytes(StandardCharsets.UTF_8));
                    messageDto.setClientId(dto.getClientId());
					ChannelCache.send2Client(robotRegisterDto.getClientId(), BeanUtil.copyProperties(messageDto,MessageDto.class));
					log.info("上线后同步数据");
				});
			} else {
				if (ChannelCache.isRegistered(dto.getClientId())) {
					if (event instanceof R<?>) {
						// 通用返回内容
						SpringUtil.getApplicationContext().publishEvent(dto);
					}  else {
						// 主动推送的消息
						SpringUtil.getApplicationContext().publishEvent(event);
					}
				} else {
					String ip = ctx.channel().remoteAddress().toString().split(":")[0].substring(1);
					log.info("未知的客户端消息断开" + ip);
					ctx.close().sync();
				}
			}
		}catch (Exception e){
			log.error("[netty handler] 消息解析失败 {}--- {}", dto.getCmd(), e.getMessage());
		}
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close().sync();
        super.exceptionCaught(ctx, cause);
    }


	/**
	 * 客户端连接异常
	 */
	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		//移除执行端相关消息
		QueryListener.removeClientMsgFromAllTopic(
				ChannelCache.getClientIdByChannelCtx(ctx)
		);
		//消息广播
		SubscriptionEndpoint.topicSubscribers.keySet().stream().forEach(
				topic -> QueryListener.broadcastMessageByTopic(topic)
		);
		ChannelCache.remove(ctx);
		ctx.close().sync();
        super.channelInactive(ctx);
    }



    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent idleEvent = (IdleStateEvent) evt;
            if (idleEvent.state() == IdleState.READER_IDLE) {
                // 读超时
                log.warn("[netty]{} read time out", ctx.channel().remoteAddress());
            } else if (idleEvent.state() == IdleState.WRITER_IDLE) {
                // 写超时
                log.warn("[netty]{} write time out", ctx.channel().remoteAddress());
                ctx.close().sync();
                ChannelCache.remove(ctx);
                super.userEventTriggered(ctx, evt);
            } else if (idleEvent.state() == IdleState.ALL_IDLE) {
                //全部
                log.warn("[netty]{} all time out", ctx.channel().remoteAddress());
                ctx.close().sync();
                ChannelCache.remove(ctx);
                super.userEventTriggered(ctx, evt);
            }
        } else {
            log.warn("[netty] {} 断开",ctx.channel().remoteAddress());
        }

    }


}
